package live.sidian.database.synchronizer.sync.data;

import cn.hutool.crypto.SecureUtil;
import live.sidian.database.synchronizer.infrastructure.sql.SqlHelper;
import live.sidian.database.synchronizer.infrastructure.sql.SqlHelper.Record;
import live.sidian.database.synchronizer.infrastructure.utils.CollUtil;
import live.sidian.database.synchronizer.sync.data.entity.Column;
import live.sidian.database.synchronizer.sync.data.entity.TablePriKey;
import live.sidian.database.synchronizer.sync.structure.model.Database;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 数据同步器
 *
 * @author sidian
 * @date 2021/4/12 23:20
 */
@Slf4j
public class DataSyncer {
    SqlHelper sourceSqlHelper;
    SqlHelper targetSqlHelper;

    /**
     * @param source 源数据库
     * @param target 目标数据库
     */
    public DataSyncer(Database source, Database target) {
        sourceSqlHelper = source.toSqlHelper();
        targetSqlHelper = target.toSqlHelper();
    }

    /**
     * 同步数据. 使用前, 请确保两者数据库结构一致.
     *
     * @param tables 要同步的表
     */
    public void sync(List<String> tables) {
        if (CollUtil.isEmpty(tables)) {
            return;
        }
        // 获取交集
        Collection<String> t = CollUtil.intersection(sourceSqlHelper.getTables(), targetSqlHelper.getTables());
        Collection<String> intersection = CollUtil.intersection(tables, t);
        // 同步
        for (String table : intersection) {
            sync(table);
        }
    }

    /**
     * 同步数据. 使用前, 请确保两者数据库结构一致.
     */
    public void sync() {
        // 获取共同存在的表
        Collection<String> tables = CollUtil.intersection(sourceSqlHelper.getTables(), targetSqlHelper.getTables());
        // 同步
        for (String table : tables) {
            sync(table);
        }
    }

    @NoArgsConstructor
    @AllArgsConstructor
    static class Diff {
        Collection<TablePriKey> willAdd;
        Collection<TablePriKey> willUpdate;
        Collection<TablePriKey> willDelete;
    }

    private void sync(String table) {
        // 对源数据库的表求md5值
        Map<TablePriKey, String> sMd5 = md5Table(table, sourceSqlHelper);
        // 对目标数据库的表求md5值
        Map<TablePriKey, String> tMd5 = md5Table(table, targetSqlHelper);
        // 计算差异
        Diff diff = diff(sMd5, tMd5);
        // 开始同步
        handle(diff);
    }

    private void handle(Diff diff) {
        handleDelete(diff.willDelete);
        handleAdd(diff.willAdd);
        handleUpdate(diff.willUpdate);
    }

    // todo 单主键时, 有优化空间
    private void handleDelete(Collection<TablePriKey> willDelete) {
        for (TablePriKey priKey : willDelete) {
            targetSqlHelper.execute(priKey.toDeleteSql(), priKey::setValue);
        }
    }

    private void handleUpdate(Collection<TablePriKey> willUpdate) {
        Map<String, Integer> columnTypeMap = null;
        boolean initial = false;
        for (TablePriKey priKey : willUpdate) {
            if (!initial) {
                initial = true;
                columnTypeMap = targetSqlHelper.getColumnTypeMap(priKey.getTable());
            }
            // 获取source的记录
            Record record = sourceSqlHelper.queryOne(priKey.toSelectSql(), priKey::setValue);
            // 修改target的记录
            String updateSql = String.format("update `%s` set %s where %s",
                    priKey.getTable(),
                    columnTypeMap.keySet().stream().map(s -> String.format("`%s` = ?", s)).collect(Collectors.joining(",")),
                    priKey.toWhereSql()
            );
            Map<String, Integer> finalColumnTypeMap = columnTypeMap;
            targetSqlHelper.execute(updateSql, statement -> {
                // 行记录
                int index = 0;
                for (Map.Entry<String, Integer> entry : finalColumnTypeMap.entrySet()) {
                    index++;
                    Object col = record.get(entry.getKey());
                    if (col == null) {
                        statement.setNull(index, entry.getValue());
                    } else {
                        statement.setObject(index, col, entry.getValue());
                    }
                }
                // 主键条件
                for (Map.Entry<String, Column> entry : priKey.entrySet()) {
                    index++;
                    Column column = entry.getValue();
                    if (column.getValue() == null) {
                        statement.setNull(index, column.getType());
                    } else {
                        statement.setObject(index, column.getValue(), column.getType());
                    }
                }
            });
        }
    }

    // todo 考虑批量添加
    private void handleAdd(Collection<TablePriKey> willAdd) {
        Map<String, Integer> columnTypeMap = null;
        boolean initial = false;
        for (TablePriKey priKey : willAdd) {
            if (!initial) {
                initial = true;
                columnTypeMap = targetSqlHelper.getColumnTypeMap(priKey.getTable());
            }
            // source的记录
            Record record = sourceSqlHelper.queryOne(priKey.toSelectSql(), priKey::setValue);
            // 添加到target中来
            String sql = String.format("insert into `%s` (%s) values (%s)",
                    priKey.getTable(),
                    columnTypeMap.keySet().stream().map(s -> String.format("`%s`", s)).collect(Collectors.joining(",")),
                    columnTypeMap.keySet().stream().map(s -> "?").collect(Collectors.joining(","))
            );
            Map<String, Integer> finalColumnTypeMap = columnTypeMap;
            targetSqlHelper.execute(sql, statement -> {
                int index = 0;
                for (Map.Entry<String, Integer> entry : finalColumnTypeMap.entrySet()) {
                    index++;
                    Object col = record.get(entry.getKey());
                    if (col == null) {
                        statement.setNull(index, entry.getValue());
                    } else {
                        statement.setObject(index, col, entry.getValue());
                    }
                }
            });
        }
    }

    private Diff diff(Map<TablePriKey, String> sMd5, Map<TablePriKey, String> tMd5) {
        // 根据主键值差分出要新增的或删除的
        Collection<TablePriKey> unknown = CollUtil.intersection(sMd5.keySet(), tMd5.keySet());
        Collection<TablePriKey> willAdd = CollUtil.subtract(sMd5.keySet(), unknown);
        Collection<TablePriKey> willDelete = CollUtil.subtract(tMd5.keySet(), unknown);
        // 根据md5差分出要修改的
        Collection<TablePriKey> willUpdate = new HashSet<>();
        unknown.forEach(tablePriKey -> {
            if (!Objects.equals(sMd5.get(tablePriKey), tMd5.get(tablePriKey))) {
                willUpdate.add(tablePriKey);
            }
        });
        return new Diff(willAdd, willUpdate, willDelete);
    }

    /**
     * 计算表的md5值
     *
     * @return 行的主键与行的md5的映射
     */
    private Map<TablePriKey, String> md5Table(String table, SqlHelper sqlHelper) {
        Map<TablePriKey, String> map = new HashMap<>();
        Map<String, Integer> typeMap = sqlHelper.getColumnTypeMap(table);
        // 获取主键的列
        Set<String> priKeyCols = sqlHelper.getPriKeyCols(table);
        // 总行数
        Long count = sqlHelper.queryOne(String.format("select count(*) as `count` from `%s`", table)).getLong("count");
        // 每次遍历100个
        for (int i = 1; i <= ((int) Math.ceil(count * 1.0 / 100)); i++) {
            List<Record> records = sqlHelper.queryList(String.format("select * from `%s` limit %d,100", table, (i - 1) * 100));
            // 计算每一行md5值
            for (Record record : records) {
                // 行的md5
                String md5 = calcRowMd5(record);
                // 主键
                TablePriKey priKey = new TablePriKey(table);
                for (String col : priKeyCols) {
                    priKey.put(col, new Column(col, typeMap.get(col), record.get(col)));
                }
                // 记录
                map.put(priKey, md5);
            }
        }
        return map;
    }

    /**
     * 计算行的md5值
     *
     * @param record 行
     * @return md5值
     */
    private String calcRowMd5(Record record) {
        // 列名自然排序
        List<String> colNames = new LinkedList<>(record.keySet());
        colNames.sort(Comparator.naturalOrder());
        // 所有列转化为字符串, 拼接
        String str = colNames.stream().map(col -> record.get(col) + "").collect(Collectors.joining(""));
        // 求md5值
        return SecureUtil.md5(str);
    }
}
